/*
 * Copyright (c) 2020-2099 the original author or authors. All rights reserve.
 */
package org.tianyun.cloud.amqp.rabbitmq.producer;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.reflect.MethodUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.util.Assert;
import org.tianyun.cloud.amqp.rabbitmq.annotation.RoutingKey;

import java.io.Serializable;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.Map;

/**
 * RabbitProducerInvocationHandler
 *
 * @auther ebert_chan
 */
public class RabbitProducerInvocationHandler implements InvocationHandler {

    private RabbitProducerFactoryBean target;

    private Map<Method, MethodHandler> dispatch;

    public RabbitProducerInvocationHandler(RabbitProducerFactoryBean target) {
        this.target = target;
        this.dispatch = new HashMap<>();
        this.resolve();
    }

    public void resolve() {
        Class<?> type = target.getType();
        Method[] methods = MethodUtils.getMethodsWithAnnotation(type, RoutingKey.class, true, false);
        if (methods == null) {
            return;
        }
        for (Method method : methods) {
            RoutingKey routingKey = method.getAnnotation(RoutingKey.class);
            String exchange = StringUtils.defaultIfEmpty(routingKey.exchange(), target.getExchange());
            Assert.hasText(exchange, "创建消息生产者失败，必须在@RoutingKey或@RabbitProducer注解上设置exchange值");
            Assert.hasText(routingKey.value(), "创建消息生产者失败，必须在@RoutingKey注解上设置value值");

            MethodHandler hander = new MethodHandler(target.getRabbitTemplate(), exchange, routingKey.value());
            hander.setApplicationContext(target.getApplicationContext());
            dispatch.put(method, hander);
        }
    }

    /*
     * @see java.lang.reflect.InvocationHandler#invoke(java.lang.Object, java.lang.reflect.Method, java.lang.Object[])
     */
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        if ("equals".equals(method.getName())) {
            try {
                Object otherHandler = args.length > 0 && args[0] != null ? args[0] : null;
                return equals(otherHandler);
            } catch (IllegalArgumentException e) {
                return false;
            }
        } else if ("hashCode".equals(method.getName())) {
            return hashCode();
        } else if ("toString".equals(method.getName())) {
            return toString();
        }

        if (dispatch.get(method) == null) {
            return null;
        }

        return dispatch.get(method).invoke(args);
    }

    @Override
    public boolean equals(Object obj) {
        if (obj instanceof RabbitProducerFactoryBean) {
            RabbitProducerFactoryBean other = (RabbitProducerFactoryBean) obj;
            return target.equals(other);
        }
        return false;
    }

    @Override
    public int hashCode() {
        return target.hashCode();
    }

    @Override
    public String toString() {
        return target.toString();
    }

    static class MethodHandler implements ApplicationContextAware {

        private ApplicationContext applicationContext;

        private String rabbitTemplate;

        private String exchange;

        private String routingKey;

        private RabbitTemplate rabbitTemplateObject;

        private MessageConverter messageConverter;

        /**
         * 构造器
         *
         * @param rabbitTemplate
         * @param exchange
         * @param routingKey
         */
        public MethodHandler(String rabbitTemplate, String exchange, String routingKey) {
            super();
            this.rabbitTemplate = rabbitTemplate;
            this.exchange = exchange;
            this.routingKey = routingKey;
        }

        private RabbitTemplate getRabbitTemplate() {
            if (rabbitTemplateObject != null) {
                return rabbitTemplateObject;
            }
            if (StringUtils.isEmpty(rabbitTemplate)) {
                rabbitTemplateObject = applicationContext.getBean(RabbitTemplate.class);
            } else {
                rabbitTemplateObject = (RabbitTemplate) applicationContext.getBean(rabbitTemplate);
            }
            return rabbitTemplateObject;
        }

        private MessageConverter getMessageConverter() {
            if (messageConverter != null) {
                return messageConverter;
            }
            if (rabbitTemplateObject == null || (messageConverter = rabbitTemplateObject.getMessageConverter()) == null) {
                messageConverter = new Jackson2JsonMessageConverter();
            }
            return messageConverter;
        }

        /*
         * @see org.springframework.context.ApplicationContextAware#setApplicationContext(org.springframework.context.ApplicationContext)
         */
        @Override
        public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
            this.applicationContext = applicationContext;
        }

        public Object invoke(Object[] argv) throws Throwable {
            if (argv == null || argv.length == 0) {
                return null;
            }

            Object object = argv[0];
            Integer delay = null;
            if (argv.length > 1) {
                if (argv[1] instanceof Integer) {
                    delay = (Integer) argv[1];
                } else if (argv[1] instanceof Long) {
                    delay = ((Long) argv[1]).intValue();
                }
            }
            RabbitTemplate rabbit = this.getRabbitTemplate();
            MessageConverter messageConverter = this.getMessageConverter();

            MessageProperties messageProperties = new MessageProperties();
            if (delay != null) {
                messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
                messageProperties.setDelay(delay);
            }

            Message message = messageConverter.toMessage(object, messageProperties);
            ReturnedMessage returnedMessage = new ReturnedMessage(message, 0, "not available", exchange, routingKey);
            CorrelationData correlationData = new CorrelationData();
            correlationData.setReturned(returnedMessage);

            rabbit.convertAndSend(exchange, routingKey, message, correlationData);
            return null;
        }

        public boolean canSerialize(Object... argv) {
            for (int i = 0; i < argv.length; i++) {
                if (argv[i] != null && !(argv[i] instanceof Serializable)) {
                    return false;
                }
            }
            return true;
        }

    }

}
